package com.niit.mock

import java.util.{Properties, Random}

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

import scala.collection.mutable.ListBuffer
//模拟用户点击广告数据：在真实情况下，这些数据会通过flume监听日志文件传递给kafka
object MockData {

    def main(args: Array[String]): Unit = {

        // 生成模拟数据
        // 格式 ：timestamp area  city  userid  adid
        // 含义： 时间戳    区域  城市 用户    广告

        // Application => Kafka => SparkStreaming => Analysis
        val prop = new Properties()
        // 添加配置
        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092")
        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
        val producer = new KafkaProducer[String, String](prop)

        while ( true ) {

            mockdata().foreach(
                data => {
                    // 向Kafka中生成数据                                主题
                    val record = new ProducerRecord[String, String]("AD2", data)
                    producer.send(record)
                    println(data)
                }
            )

            Thread.sleep(2000)
        }

    }
    def mockdata() = {
        val list = ListBuffer[String]() //用来生成的模拟数据
        val areaList = ListBuffer[String]("华北", "华东", "华南")//模拟区域
        val cityList = ListBuffer[String]("北京", "上海", "深圳")//模拟城市

        for ( i <- 1 to new Random().nextInt(50) ) {//生成1 -50个随机

            val area = areaList(new Random().nextInt(3))//华东
            val city = cityList(new Random().nextInt(3))//北京
            var userid = new Random().nextInt(6) + 1 //模拟生成用户id 1-6
            var adid = new Random().nextInt(6) + 1   //模拟生成广告id 1-6
            //拼接模拟数据进行保存
            list.append(s"${System.currentTimeMillis()} ${area} ${city} ${userid} ${adid}")
        }
        //返回模拟数据
        list
    }

}
